package com.rem.flink.flink4Sink;

import cn.hutool.json.JSONUtil;
import com.rem.flink.flink2Source.ClickSource;
import com.rem.flink.flink2Source.Event;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.redisson.Redisson;
import org.redisson.api.RMap;
import org.redisson.api.RedissonClient;
import org.redisson.codec.TypedJsonJacksonCodec;
import org.redisson.config.Config;

import java.util.List;

/**
 * 自定义 sink 通过redisson 输出到redis
 *
 * @author Rem
 * @date 2022-10-10
 */

public class SinkToCustomer {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> stream = env.addSource(new ClickSource());

        stream.addSink(new RichSinkFunction<>() {

            private RedissonClient redissonClient;

            @Override
            public void open(Configuration parameters) {
                Config config = new Config();
                config.setCodec(new TypedJsonJacksonCodec(Event.class));
                config.useSentinelServers()
                        .setMasterName("mymaster")
                        .setPassword("123456")
                        .setSentinelAddresses(List.of("redis://localhost:6380", "redis://localhost:6381", "redis://localhost:6382"));
                this.redissonClient = Redisson.create(config);
            }


            @Override
            public void invoke(Event event, Context context) throws Exception {


                RMap<Object, Object> clicks3 = redissonClient.getMap("clicks5");
                String s = JSONUtil.toJsonStr(event);
                clicks3.put(event.getTimestamp(), event);
            }

            @Override
            public void close() throws Exception {
                if (redissonClient != null) {
                    redissonClient.shutdown();
                }
            }
        });

        env.execute();
    }
}

